package cn.hinglo.common.plugin.mqtt;

import cn.hinglo.common.util.PropertiesKit;
import cn.hutool.core.util.ClassUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * @createTime: 2020/3/30 : 11:07
 * @author: HingLo
 * @description: mtqq消息管理中间件
 */
@Data
@Slf4j
public class MqttManager {
    /***
     * 链接地址
     */
    public static final String HOST = PropertiesKit.get("mtqq.host", "tcp://127.0.0.1:1883");
    /***
     * 客户端唯一ID
     */
    private static final String CLIENT_ID = PropertiesKit.get("mtqq.clientId", IdUtil.objectId());

    /***
     * 是否清理回话
     */
    private Boolean cleanSession = PropertiesKit.getToBool("mtqq.cleanSession");
    /***
     * 链接用户名
     */
    private String username = PropertiesKit.get("mtqq.username");
    /***
     * 链接密码
     */
    private String password = PropertiesKit.get("mtqq.password", "");
    /***
     * 链接超时 单位：s
     */
    private Integer timeout = PropertiesKit.getToInt("mtqq.timeOut", 30);
    /***
     * 心跳检测
     */
    private Integer keepAliveInterval = PropertiesKit.getToInt("mtqq.keepAliveInterval", 60);

    /***
     * 设置是否自动重连，默认true
     */
    private Boolean automaticReconnect = PropertiesKit.getToBool("mtqq.automaticReconnect", true);

    /***
     * 是否开启自动描述注解实现类
     */
    private boolean scan = PropertiesKit.getToBool("mtqq.scan", true);
    /***
     * 设置自动扫描包的字符串
     */
    private String packageName = PropertiesKit.get("mtqq.packageName", "cn.hinglo.common.plugin.mqtt");


    /***
     * 客户端链接对象
     */
    private MqttClient client;
    /***
     * 链接属性对象
     */
    private MqttConnectOptions options;


    /***
     * 所有需要监听的主题
     */
    private List<MqttDto> mqttDtoList = new ArrayList<>();

    private final static MqttManager MQTT_MANAGER = new MqttManager();

    private MqttManager() {
    }

    public static MqttManager getSingleton() {
        return MQTT_MANAGER;
    }


    public boolean start() {
        try {
            // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(HOST, CLIENT_ID, new MemoryPersistence());
            // MQTT的连接设置
            options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(cleanSession);
            // 设置连接的用户名
            options.setUserName(username);
            // 设置连接的密码
            options.setPassword(password.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(timeout);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(keepAliveInterval);
            // 设置是否断开重连
            options.setAutomaticReconnect(automaticReconnect);
            client.connect(options);
            // 注解扫描
            if (scan) {
                if (StrUtil.isBlank(this.packageName)) {
                    throw new RuntimeException("自动扫描Mqtt 消息处理器的包路径不能为空");
                }
                this.scan(packageName);
            }
            String[] topicFilters = new String[mqttDtoList.size()];
            int[] qos = new int[mqttDtoList.size()];
            IMqttMessageListener[] messageListeners = new IMqttMessageListener[mqttDtoList.size()];
            for (int i = 0, len = mqttDtoList.size(); i < len; i++) {
                topicFilters[i] = mqttDtoList.get(i).getTopicName();
                qos[i] = mqttDtoList.get(i).getQos();
                messageListeners[i] = mqttDtoList.get(i).getIMqttMessageListener();
            }
            client.subscribe(topicFilters, qos, messageListeners);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("MTQQ 启动失败", e);
            return false;
        }
        return true;
    }


    /***
     * 注解扫描监听器
     * @param scanPackage 定义的包
     */
    @SneakyThrows
    public void scan(String scanPackage) {
        // 第一步,将获取包下面所有的class
        Set<Class<?>> classList = ClassUtil.scanPackageByAnnotation(scanPackage, MqttListener.class);
        Class<IMqttMessageListener> aClass = IMqttMessageListener.class;
        // 第二步,草扫描带有MtqqListener注解的类
        for (Class<?> className : classList) {
            if (!aClass.isAssignableFrom(className) || aClass.equals(className)) {
                continue;
            }
            try {
                MqttListener msgListener = className.getAnnotation(MqttListener.class);
                IMqttMessageListener messageListener = (IMqttMessageListener) className.newInstance();
                MqttDto mqttDto = new MqttDto(msgListener.value(), msgListener.qos(), messageListener);
                mqttDtoList.add(mqttDto);
            } catch (IllegalAccessException | InstantiationException e) {
                e.printStackTrace();
            }
        }
    }


    @SneakyThrows
    public boolean stop() {
        client.disconnect();
        client.close();
        return true;
    }


    @Data
    @AllArgsConstructor
    class MqttDto {
        /***
         * 主题名称
         */
        private String topicName;

        /***
         * 消息响应类型
         */
        private int qos;

        /***
         * 消息处理器
         */
        private IMqttMessageListener iMqttMessageListener;

    }

    public static void main(String[] args) {
        MqttManager client = MqttManager.getSingleton();
        client.start();
    }


}
